package com.demo.common.redis.stream;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;

/**
 * <p>
 * 自定义分组消费者监听
 * </p>
 *
 * @author molong
 * @date 2022/6/22
 */
@Slf4j
public abstract class GroupStreamListener<T> extends BaseStreamListener<T> {

    protected GroupStreamListener() {
        super.setReadOffset(ReadOffset.lastConsumed());
    }

    /**
     * 获取一个消费组信息，包含组名称和消费者名称
     * @return 消费组
     */
    public abstract Consumer getConsumer();


    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        try {
            log.info("---key：{}, 接收到消息：messageId = {}", getFullStreamKey(), message.getId());
            T t = super.deserialization(message);
            onMessage(t);
            //如果没有报错，将消息设置为已收到
            Long acknowledge = getRedisTemplate().opsForStream().acknowledge(getConsumer().getGroup(), message);
            log.info("----key: {}, 消息处理完毕：messageId={}, 确认收到消息的长度：{}", getFullStreamKey(), message.getId(), acknowledge);
            //执行回调函数
            callback(message.getId());
        } catch (Exception e){
            exceptionHandler(e, message.getId().getValue(), message);
        }
    }

}
